package com.gui.migrate;

import com.gui.connect.Command;
import com.gui.connect.Connector;
import com.gui.connect.Processor;
import com.gui.connect.RedisReply;
import com.gui.exceptions.RedisConnectionException;
import com.gui.util.Cmd;
import com.gui.util.ReplyType;
import com.gui.util.SafeEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Collections;
import java.util.LinkedList;
import java.util.Queue;

/**
 * Created by Gui on 2017/7/23.
 */
public class ReadContext implements Closeable {

    private static final Logger logger = LoggerFactory.getLogger(ReadContext.class);

    private Context context;

    private Connector connector;
    private Command command;
    private Processor processor;

    public ReadContext(Context context) {
        this.context = context;
        connector = new Connector(context.sourceHost, context.sourcePort);
    }

    public void start() throws IOException {
        connectSourceRedis();
        if (!beginReplication()) {
            logger.error("Redis replication failed");
        }
    }

    public void stop() throws IOException {
        connector.close();
    }

    private void connectSourceRedis() throws IOException {
        logger.info("Connecting source redis : {}:{}", context.sourceHost, context.sourcePort);

        connector.connect();
        processor = connector.getProcessor();
        command = connector.getCommand();
    }

    private boolean beginReplication() throws IOException {
        logger.info("Begin replication");

        RedisReply reply = null;

        //PING
        command.sendCmd(Cmd.PING);
        reply = processor.getObjectReply();

        if (reply.getType() == ReplyType.ERROR && reply.getReply().equals("NOAUTH")) {
            //AUTH
            command.sendCmd(Cmd.AUTH, context.sourcePasswd);
            reply = processor.getObjectReply();
            if (reply.getType() == ReplyType.ERROR) {
                logger.error("error password for source redis");
                return false;
            }
        }

        String lp = context.listeningPort >= 0 ? String.valueOf(context.listeningPort) : "12345";
        logger.warn("REPLCONF listening-port : {}", lp);

        command.sendBytesCRLF(SafeEncoder.encode("REPLCONF listening-port " + lp));
        reply = processor.getObjectReply();
        if (reply.getType() == ReplyType.ERROR) {
            logger.error("REPLCONF listening-port error");
            return false;
        }

        command.sendBytesCRLF(SafeEncoder.encode("REPLCONF capa eof"));
        reply = processor.getObjectReply();
        if (reply.getType() == ReplyType.ERROR) {
            logger.error("REPLCONF capa eof error");
            return false;
        }

        String psyncStr = null;
        if (context.masterRunid == null || context.masterOffset <= 0) {
            psyncStr = "PSYNC ? -1";
        } else {
            psyncStr = "PSYNC " + context.masterRunid + " " + context.masterOffset;
        }
        command.sendBytesCRLF(SafeEncoder.encode(psyncStr));
        reply = processor.getObjectReply();

        if (reply.getType() == ReplyType.STATUS) {
            if (reply.getReply().equals("CONTINUE")) {
                return this.beginLoadAOF();

            } else if (((String)reply.getReply()).startsWith("FULLRESYNC")) {
                System.out.println(reply.getReply());
                String[] strs = reply.getReply().toString().split(" ");
                context.masterRunid = strs[1];
                context.masterOffset = context.slaveOffset = Long.parseLong(strs[2]);

                if (beginLoadRDB()) {
                    return this.beginLoadAOF();
                }

            } else {
                logger.error("Unknow PSYNC reply: ", reply.getReply());
                return false;
            }
        } else {
            logger.error("PSYNC error: {}", reply.getReply());
            return false;
        }

        return true;
    }

    private boolean beginLoadRDB() {
        context.startReadRDB();

        File file = null;
        FileOutputStream fos = null;
        try {
            String fileName = String.format("node-%s-%d-%d.rdb", context.sourceHost, context.sourcePort, System.currentTimeMillis());
            file = new File(context.rdbPath, fileName);
            fos = new FileOutputStream(file);
            processor.saveRDBData(fos);
        } catch (IOException e) {
            logger.error(e.toString());
            return false;
        } finally {
            if (fos != null) {
                try {
                    fos.close();
                } catch (IOException e) {
                    logger.error(e.toString());
                }
            }
        }
        context.finishReadRDB(file);

        return true;
    }

    private boolean beginLoadAOF() {
        context.startReadAOF();

        while (true) {
            byte[] aof = null;
            try {
                aof = processor.loadSingleAOF();
            } catch (IOException e) {
                logger.error(e.toString());
                context.brokeReadAOF();
                return false;
            }
            if (aof == null || aof.length <= 0) {
                continue;
            }

            context.masterOffset += aof.length;
            context.aofQueue.add(aof);

            String str = SafeEncoder.decode(aof);
            System.out.println(str);
        }
    }

    public void replyAckToMaster() {
        String ack = "REPLCONF ACK " + context.masterOffset;
        System.out.println(ack);
        command.sendBytesCRLF(SafeEncoder.encode(ack));
    }

    @Override
    public void close() throws IOException {
        connector.close();
    }
}
